package com.dimld.connect.tcp;

import com.dimld.connect.ConnectContainer;
import com.dimld.connect.ConnectorI;
import com.dimld.connect.MessageStructure;
import com.dimld.connect.MsgReaderI;
import com.dimld.utils.Tools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.Serializable;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class TcpClient implements Runnable, ConnectorI, Serializable {
    private transient Logger                log          = LoggerFactory.getLogger("tcp");
    private transient MsgReaderI            msg_reader;
    private transient DataInputStream       din          = null;
    private transient Thread                senderThread = null;
    private transient Thread                readerThread = null;
    private transient BlockingQueue<byte[]> readerQueue  = null;
    private transient Socket                socket;
    private transient Object                lock;
    private           String                bindServiceUuid;

    private Object                                data;
    private String                                uuid        = UUID.randomUUID().toString();
    private String                                fullIp;
    private boolean                               isConnected = false;
    private long                                  createTime  = System.currentTimeMillis();
    private boolean                               running     = true;
    private boolean                               threadReady = false;
    private boolean                               senderReady = false;
    private LinkedBlockingQueue<MessageStructure> msgQueue    = new LinkedBlockingQueue<>();
//    public IQueue<MessageStructure> getQueue() {
//        return MainEntry.cluster.getQueue(ClusterConstants.tcpQueue_ + getUuid());
//    }

    public TcpClient() {
    }

    public TcpClient(Socket socket, MsgReaderI msg_reader, boolean is_server, int timeout, String serviceUUID) {
        this.msg_reader = msg_reader;
        this.bindServiceUuid = serviceUUID;
        this.socket = socket;
        this.fullIp = socket.getRemoteSocketAddress().toString().replace("/", "");
        if(is_server) {
            try {
                this.socket.setSoTimeout(timeout * 1000);
            } catch(SocketException e) {
                this.log.error("[TCP][" + this.fullIp + "][" + this.uuid + "]连接超时", e);
            }
        } else {
            this.log.info("[TCP][" + this.fullIp + "][" + this.uuid + "]客户端方式启动");
        }
        this.log.info("[TCP][" + this.fullIp + "][" + this.uuid + "]创建连接对象，等待执行线程");
    }

    @Override
    public void run() {
        Thread.currentThread().setName("TcpClient_" + getUuid());
        String logPrefix = "[TCP][" + this.fullIp + "][" + this.uuid + "]";
        try {
            this.log.info(logPrefix + "连接成功，监听消息线程启动");
            this.isConnected = true;
            this.din = new DataInputStream(this.socket.getInputStream());
            ClientMsgSender sender = new ClientMsgSender(this);
            this.senderThread = new Thread(sender);
            this.senderThread.start();
            ClientMsgReader reader = new ClientMsgReader(this);
            this.readerThread = new Thread(reader);
            this.log.info(logPrefix + "开始接收消息线程");
            this.readerThread.start();
            ConnectContainer.put(this);
            threadReady = true;
            ready();
            while(this.running) {
                byte[] dataBody = receive();
                if(dataBody != null) {
                    this.readerQueue.add(dataBody);
                } else
                    close("[TCP]收到的消息为null，移除此连接");
            }
        } catch(SocketException e) {
            this.log.error(logPrefix + "Socket异常断开", e);
        } catch(IOException e) {
            this.log.error(logPrefix + "Socket读取数据异常", e);
        } catch(Exception e) {
            this.log.error(logPrefix + "其他异常", e);
        } finally {
            close(logPrefix + "主线程结束");
            this.readerQueue = null;
            this.socket = null;
        }
        ConnectContainer.remove(this.uuid);
        if(this.msg_reader != null) {
            this.msg_reader.disconnect(this);
        }
        destroy();
        this.log.info("[TCP][" + this.fullIp + "][" + this.uuid + "]连接断开，主线程结束");
    }

    private void ready() {
        if(this.msg_reader != null) {
            if(readerQueue != null && senderReady && threadReady) {
                log.info("[TCP]TCP初始化完成!");
                this.msg_reader.connect(this);
            } else {
                if(readerQueue == null) {
                    log.info("[TCP]readerQueue 没有准备就绪");
                }
                if(!senderReady) {
                    log.info("[TCP]senderQueue 没有准备就绪");
                }
                if(!threadReady) {
                    log.info("[TCP]线程没有准备就绪");
                }
            }
        } else {
            log.info("[TCP]没有实现消息接口");
        }
    }

    private byte[] receive() {
        int    msgLength;
        byte[] resp = null;
        try {
            msgLength = this.din.readInt();
            int realSize = 0;
            if(msgLength > 4) {
                // int sequence = din.readInt();
                int trueSize = msgLength - 4;
                // this.log.info("[TCP]接收消息，消息序号[" + sequence + "] 消息长度[" +
                // msgLength + "]");
                if(trueSize > 1024 * 1024) {
                    close("接收的数据长度不能大于1M");
                } else {
                    resp = new byte[trueSize];
                    try {
                        do {
                            realSize += this.din.read(resp, realSize, trueSize - realSize);
                        } while(realSize != trueSize);
                    } catch(Exception e) {
                        this.log.error("[TCP][" + this.fullIp + "][" + this.uuid + "]收包异常", e);
                        close("收包异常");
                    }
                }
            } else {
                this.log.error("[TCP][" + this.fullIp + "][" + this.uuid + "]读取长度异常，长度(" + msgLength + ")。");
            }
            this.log.info("[TCP][" + this.fullIp + "][" + this.uuid + "]发来消息，包大小:" + msgLength);
        } catch(EOFException | SocketException e) {
            this.log.info("[TCP][" + this.fullIp + "][" + this.uuid + "]关闭连接");
        } catch(SocketTimeoutException e) {
            this.log.info("[TCP][" + this.fullIp + "][" + this.uuid + "]连接超时");
            close("连接超时");
        } catch(Exception e) {
            this.log.error("[TCP][" + this.fullIp + "][" + this.uuid + "]其他异常" + e.getMessage(), e);
        }
        return resp;
    }

    @Override
    public void send(String content) {
        try {
            if(ConnectContainer.get(this.uuid) != null) {
                msgQueue.put(new MessageStructure(MessageStructure.SEND, content));
            }
        } catch(InterruptedException e) {
            log.error("被中断", e);
        }
    }

    @Override
    public String getConnType() {
        return "TCP";
    }

    @Override
    public String getUuid() {
        return this.uuid;
    }

    @Override
    public MsgReaderI getReader() {
        return this.msg_reader;
    }

    @Override
    public void setData(Object data) {
        this.data = data;
    }

    @Override
    public Object getData() {
        return this.data;
    }

    @Override
    public boolean isConnected() {
        return isConnected;
    }

    @Override
    public String getFullIP() {
        return this.fullIp;
    }

    @Override
    public long getCreateTimestamp() {
        return this.createTime;
    }

    @Override
    public void close() {
        try {
            if(ConnectContainer.get(this.uuid) != null) {
                log.debug("[TCP][" + getFullIP() + "]添加关闭连接消息到队列");
                msgQueue.put(new MessageStructure(MessageStructure.CLOSE, null));
            }
        } catch(InterruptedException e) {
            log.error("被中断");
        }
    }

    public void destroy() {
        msgQueue.clear();
    }

    public String getShortIP() {
        return socket.getInetAddress().toString().replace("/", "");
    }

    public void setLock(Object lock) {
        this.lock = lock;
    }

    public Object getLock() {
        return this.lock;
    }

    private void close(String reason) {
        String logPrefix = "[TCP][" + this.fullIp + "][" + this.uuid + "]";
        isConnected = false;
        this.running = false;
        if((this.senderThread != null) && (this.senderThread.isAlive())) {
            this.senderThread.interrupt();
        }
        if((this.readerThread != null) && (this.readerThread.isAlive())) {
            this.readerThread.interrupt();
        }
        this.log.info(logPrefix + "请求关闭socket连接--" + reason);
        if(this.socket != null)
            try {
                if(!this.socket.isClosed()) {
                    this.log.info(logPrefix + "关闭socket连接");
                    this.socket.close();
                } else {
                    this.log.info(logPrefix + "socket已经被关闭");
                }
            } catch(Exception e) {
                this.log.error(logPrefix + "关闭socket异常", e);
            }
        else {
            this.log.info(logPrefix + "socket为null");
        }
        ConnectContainer.put(this);
    }

    @Override
    public String getBindServiceUuid() {
        return bindServiceUuid;
    }

    public void setBindServiceUuid(String bindServiceUuid) {
        this.bindServiceUuid = bindServiceUuid;
    }

    class ClientMsgReader implements Runnable {
        TcpClient tcpClient;

        ClientMsgReader(TcpClient tcpClient) {
            this.tcpClient = tcpClient;
        }

        public void run() {
            Thread.currentThread().setName("ClientMsgReader_" + getUuid());
            String logPrefix = "[TCP][" + fullIp + "][" + uuid + "]";
            log.info(logPrefix + "读取消息队列启动");
            byte[] waitData = null;
            readerQueue = new LinkedBlockingQueue<>();
            ready();
            while(running) {
                try {
                    if(readerQueue != null)
                        waitData = readerQueue.take();
                } catch(InterruptedException e) {
                    log.info(logPrefix + "读取消息队列被中断");
                    break;
                } catch(Exception e) {
                    log.error(logPrefix + "读取消息队列异常", e);
                    break;
                }
                if(waitData != null) {
                    try {
                        String bodyStr = new String(waitData, StandardCharsets.UTF_8);
                        log.info(logPrefix + "读取消息队列收到消息：" + bodyStr);
                        if(msg_reader != null) {
                            msg_reader.read(bodyStr, tcpClient);
                        }
                        if(readerQueue != null)
                            log.info(logPrefix + "读取消息队列剩余数量:" + readerQueue.size());
                        else
                            log.info(logPrefix + "读取消息队列已被移除");
                    } catch(Exception e) {
                        log.error(logPrefix + "读取消息队列异常", e);
                    }
                }
            }
            log.info(logPrefix + "读取消息队列结束");
            close("读取消息队列结束，尝试关闭");
        }
    }

    class ClientMsgSender implements Runnable {
        TcpClient tcpClient;

        ClientMsgSender(TcpClient tcpClient) {
            this.tcpClient = tcpClient;
        }

        public void run() {
            Thread.currentThread().setName("ClientMsgSender_" + getUuid());
            String logPrefix = "[TCP][" + fullIp + "][" + uuid + "]";
            log.info(logPrefix + "发送消息队列启动");
            senderReady = true;
            ready();
//            IQueue<MessageStructure> queue = getQueue();
            while(running) {
                try {
                    MessageStructure messageStructure = msgQueue.take();
                    if(messageStructure.getType() == 1) {
                        String message = messageStructure.getMessage();
                        log.info(logPrefix + "发送消息：\n" + message);
                        byte[] body    = message.getBytes(StandardCharsets.UTF_8);
                        byte[] re_data = new byte[4 + body.length];
                        System.arraycopy(Tools.int2byte(4 + body.length), 0, re_data, 0, 4);
                        // System.arraycopy(Tools.int2byte(sequence.getAndIncrement()), 0, re_data, 4, 4);
                        System.arraycopy(body, 0, re_data, 4, body.length);
                        if((socket == null) || (socket.isClosed())) {
                            log.info(logPrefix + "socket已经关闭");
                        } else {
                            try {
                                socket.getOutputStream().write(re_data);
                                socket.getOutputStream().flush();
                            } catch(SocketException e) {
                                if(e.getMessage().contains("reset"))
                                    log.info(logPrefix + "远端socket已经重置 " + e.getMessage());
                                else if(e.getMessage().contains("Broken"))
                                    log.info(logPrefix + "远端socket已经关闭 " + e.getMessage());
                                else if(e.getMessage().contains("closed"))
                                    log.info(logPrefix + "远端socket已经关闭 " + e.getMessage());
                                else
                                    log.error(logPrefix + "Socket发送消息异常 " + e.getMessage(), e);
                            } catch(IOException e) {
                                log.info(logPrefix + "远端socket已经关闭 " + e.getMessage());
                                break;
                            }
                            log.info(logPrefix + "发送消息队列剩余数量:" + msgQueue.size());
                        }
                    } else {
                        close("请求关闭");
                    }
                } catch(InterruptedException e) {
                    log.info(logPrefix + "发送消息队列被中断");
                } catch(Exception e) {
                    log.error(logPrefix + "发送消息队列异常", e);
                    break;
                }
            }
            log.info(logPrefix + "发送消息队列结束");
            close("发送消息队列结束，尝试关闭");
        }
    }
}